package com.binance.zkconfig.app.util;

import com.binance.zkconfig.app.constant.Constant;
import com.binance.zkconfig.app.listener.event.ZkConfigChangeEvent;
import com.binance.zkconfig.app.service.ZkConfig;
import com.binance.zkconfig.app.service.impl.ZkConfigImpl;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.nio.charset.Charset;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_ADDED;
import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_REMOVED;
import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_UPDATED;

/**
 * Created by Fei.Huang on 2018/8/28.
 */
@Slf4j
public final class ZkConfigUtils {

    private static final int MAX_RETRIES = 10;
    private static final int BASE_SLEEP_TIMES = 1000;
    private static final int MAX_WAIT_TIME_IN_SECONDS = 10;

    private static ThreadLocal<String> zookeeperUrlLocal = new ThreadLocal<>();
    private static ThreadLocal<CuratorFramework> curatorLocal = new ThreadLocal<>();

    private static final Map<String, ZkConfig> CONFIGS = Maps.newConcurrentMap();

    public static CuratorFramework getCuratorFramework(String zkUrl) {

        CuratorFramework curatorFramework = curatorLocal.get();
        if (null != curatorFramework) {
            return curatorFramework;
        }

        try {
            String connectString;
            if (StringUtils.isNotBlank(zkUrl)) {
                connectString = zkUrl;
            } else {
                connectString = zookeeperUrlLocal.get();
            }
            if (StringUtils.isBlank(connectString)) {
                throw new IllegalArgumentException("ZkConfig 'zookeeper.url' is missing");
            }

            RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIMES, MAX_RETRIES);
            curatorFramework = CuratorFrameworkFactory.builder().connectString(connectString).retryPolicy(retryPolicy).build();
            curatorFramework.start();

            log.info("ZkConfig initializing curatorFramework for ZkConfigUtils");
            curatorFramework.blockUntilConnected(MAX_WAIT_TIME_IN_SECONDS, TimeUnit.SECONDS);

            curatorLocal.set(curatorFramework);
            log.info("ZkConfig connected to zookeeper");

        } catch (Exception e) {
            throw new IllegalStateException("ZkConfig initialize curatorFramework for ZkConfigUtils failed", e);
        }

        return curatorFramework;
    }


    public static ZkConfig getConfig(String namespace) {
        ZkConfig config = CONFIGS.get(namespace);
        if (null == config) {
            synchronized (ZkConfigUtils.class) {
                config = CONFIGS.get(namespace);
                if (null == config) {
                    config = new ZkConfigImpl(namespace);
                    CONFIGS.put(namespace, config);
                }
            }
        }
        return config;
    }

    private static final Charset UTF8 = Charset.forName(Constant.UTF_8);

    public static String sanitizeKey(String context, String path) {
        return path.replace(context + Constant.SLASH, "").replace('/', '.');
    }

    public static ZkConfigChangeEvent.Type transformType(TreeCacheEvent.Type eventType) {
        if (eventType == NODE_ADDED) {
            return ZkConfigChangeEvent.Type.ADD;
        } else if (eventType == NODE_UPDATED) {
            return ZkConfigChangeEvent.Type.UPDATE;
        } else if (eventType == NODE_REMOVED) {
            return ZkConfigChangeEvent.Type.REMOVE;
        }
        return null;
    }

    public static String getEventDesc(TreeCacheEvent event, boolean sensitive) {
        StringBuilder out = new StringBuilder();
        out.append("type=").append(event.getType());
        if (event.getData() == null) {
            return out.toString();
        }
        out.append(", path=").append(event.getData().getPath());
        byte[] data = event.getData().getData();
        if (data != null && data.length > 0) {
            out.append(", data=").append(sensitive ? "******" : new String(data, UTF8));
        }
        return out.toString();
    }
}